#include "appkit/async.h"

#include <sys/prctl.h>

#include "appkit/strutil.h"
#include "appkit/tracer.h"

namespace appkit {

AsyncTaskQueue::AsyncTaskQueue(const std::string& name) : m_name(name) {}

AsyncTaskQueue::~AsyncTaskQueue() {
    if (m_startFlag.exchange(false)) {
        for (auto i = 0; i < m_workerNum; i++) {
            if (m_workers[i].valid()) {
                m_workers[i].wait();
            }
        }
    }
}

bool AsyncTaskQueue::init(int workers) {
    if (m_startFlag.load()) {
        return true;
    }
    if (workers <= 0) {
        TRACE_ERR_CLASS("invalid worker nums: %d", workers);
        return false;
    }
    m_startFlag.store(true);
    for (auto i = 0; i < workers; ++i) {  // 创建工作线程
        auto future = std::async(std::launch::async, [&]() {
            if (!m_name.empty()) {
                auto name =
                    StrUtil::format("%s_%x", m_name.data(), Thread::threadID());
                prctl(PR_SET_NAME, name.data());  // 设置线程名称
            }
            while (m_startFlag.load()) {
                auto task = getTask();
                if (task) {
                    task();
                }
            }
        });
        m_workers.push_back(std::move(future));
    }
    m_workerNum = workers;
    return true;
}

void AsyncTaskQueue::putTask(const AsyncTaskQueue::Task& task) {
    {
        std::lock_guard<std::mutex> lock(m_cvMutex);
        m_taskQueue.emplace(task);
    }
    m_cond.notify_one();
}

uint32 AsyncTaskQueue::size() {
    std::lock_guard<std::mutex> lock(m_cvMutex);
    return m_taskQueue.size();
}

AsyncTaskQueue::Task AsyncTaskQueue::getTask() {
    Task task = nullptr;
    std::unique_lock<std::mutex> lock(m_cvMutex);
    // wait_for在超时并且predic值为false时返回false,其他情况下返回true
    if (!m_cond.wait_for(lock, std::chrono::milliseconds(10),
                         [&] { return m_taskQueue.size() != 0; })) {
        return task;
    }
    if (m_taskQueue.size() != 0) {
        task = m_taskQueue.front();
        m_taskQueue.pop();
    }
    return task;
}

}  // namespace appkit
